package com.xiguanlezz.cn.common.remote.client.grpc;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.xiguanlezz.cn.api.grpc.auto.Payload;
import com.xiguanlezz.cn.api.grpc.auto.RequestGrpc;
import com.xiguanlezz.cn.api.remote.RequestCallBack;
import com.xiguanlezz.cn.api.remote.RequestFuture;
import com.xiguanlezz.cn.api.remote.request.Request;
import com.xiguanlezz.cn.api.remote.response.Response;
import com.xiguanlezz.cn.common.remote.client.Connection;
import com.xiguanlezz.cn.common.remote.client.RpcClient;
import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;
import lombok.Setter;

import java.util.Objects;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * @Author：习惯了种种
 * @CreateDate：2024/11/19 15:11
 * @Description：GRPC发送请求类
 */
@Setter
public class GrpcConnection extends Connection {
    private ManagedChannel channel;
    private Executor executor;

    protected RequestGrpc.RequestFutureStub grpcNormalServiceStub; /* 客户端到服务端单向通信代理对象 */
    protected StreamObserver<Payload> biStreamObserver; /* 客户端和服务端双向流式通信通道 */

    public GrpcConnection(RpcClient.ServerInfo serverInfo) {
        super(serverInfo);
    }

    // 单向发请求
    @Override
    public Response request(Request request, long timeoutMills) throws Exception {
        ListenableFuture<Payload> requestFuture = grpcNormalServiceStub.request(GrpcUtils.convert(request));
        Payload responsePayload = null;
        if (timeoutMills <= 0) {
            responsePayload = requestFuture.get();
        } else {
            responsePayload = requestFuture.get(timeoutMills, TimeUnit.MILLISECONDS);
        }
        return (Response) GrpcUtils.parse(responsePayload);
    }

    // 单向发请求
    @Override
    public RequestFuture requestFuture(Request request) throws Exception {
        ListenableFuture<Payload> requestFuture = grpcNormalServiceStub.request(GrpcUtils.convert(request));
        return new RequestFuture() {
            @Override
            public boolean isDone() {
                return requestFuture.isDone();
            }

            @Override
            public Response get() throws Exception {
                Payload responsePayload = requestFuture.get();
                return (Response) GrpcUtils.parse(responsePayload);
            }

            @Override
            public Response get(long timeout) throws Exception {
                Payload responsePayload = requestFuture.get(timeout, TimeUnit.MILLISECONDS);
                return (Response) GrpcUtils.parse(responsePayload);
            }
        };
    }

    // 单向发请求
    @Override
    public void asyncRequest(Request request, RequestCallBack requestCallBack) throws Exception {
        ListenableFuture<Payload> requestFuture = grpcNormalServiceStub.request(GrpcUtils.convert(request));
        // Futures.addCallback第三个参数为执行回调函数的线程池
        Futures.addCallback(requestFuture, new FutureCallback<Payload>() {
            @Override
            public void onSuccess(Payload responsePayload) {
                Response response = (Response) GrpcUtils.parse(responsePayload);
                requestCallBack.getExecutor().execute(() -> {
                    requestCallBack.onResponse(response);
                });
            }

            @Override
            public void onFailure(Throwable t) {
                requestCallBack.onException(t);
            }
        }, Objects.nonNull(requestCallBack.getExecutor()) ? requestCallBack.getExecutor() : executor);
    }

    // 双向发请求
    public void sendResponse(Response response) {
        Payload convert = GrpcUtils.convert(response);
        biStreamObserver.onNext(convert);
    }

    // 双向发请求
    public void sendRequest(Request request) {
        Payload convert = GrpcUtils.convert(request);
        biStreamObserver.onNext(convert);
    }

    public void close() {
        if (Objects.nonNull(biStreamObserver)) {
            try {
                biStreamObserver.onCompleted();
            } catch (Exception e) {

            }
        }
        if (Objects.nonNull(channel) && !channel.isShutdown()) {
            try {
                channel.shutdownNow();
            } catch (Exception e) {
            }
        }
    }

}
